package com.ctsi.commons.util.async;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * 异步任务返回结果
 * 
 * @author yueming
 *
 * @param <V>
 */
public class TaskFuture<V> implements ListenableFuture<V> {
	private static final Logger log = LoggerFactory.getLogger(TaskFuture.class);
	private static final CauseHolder CANCELLATION_CAUSE_HOLDER = new CauseHolder(new CancellationException());

	static {
		CANCELLATION_CAUSE_HOLDER.cause.setStackTrace(new StackTraceElement[0]);
	}

	private volatile Object result;

	private short waiters;

	private boolean hasWaiters() {
		return waiters > 0;
	}

	private void incWaiters() {
		if (waiters == Short.MAX_VALUE) {
			throw new IllegalStateException("too many waiters: " + this);
		}
		waiters++;
	}

	private void decWaiters() {
		waiters--;
	}

	private boolean await0(long timeoutNanos, boolean interruptable) throws InterruptedException {
		if (isDone()) {
			return true;
		}

		if (timeoutNanos <= 0) {
			return isDone();
		}

		if (interruptable && Thread.interrupted()) {
			throw new InterruptedException(toString());
		}

		long startTime = System.nanoTime();
		long waitTime = timeoutNanos;
		boolean interrupted = false;

		try {
			synchronized (this) {
				if (isDone()) {
					return true;
				}

				if (waitTime <= 0) {
					return isDone();
				}

				incWaiters();
				try {
					for (;;) {
						try {
							wait(waitTime / 1000000, (int) (waitTime % 1000000));
						} catch (InterruptedException e) {
							if (interruptable) {
								throw e;
							} else {
								interrupted = true;
							}
						}

						if (isDone()) {
							return true;
						} else {
							waitTime = timeoutNanos - (System.nanoTime() - startTime);
							if (waitTime <= 0) {
								return isDone();
							}
						}
					}
				} finally {
					decWaiters();
				}
			}
		} finally {
			if (interrupted) {
				Thread.currentThread().interrupt();
			}
		}
	}

	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		Object result = this.result;
		if (isDone0(result) || result == TaskState.UNCANCELLABLE) {
			return false;
		}

		synchronized (this) {
			// Allow only once.
			result = this.result;
			if (isDone0(result) || result == TaskState.UNCANCELLABLE) {
				return false;
			}

			this.result = CANCELLATION_CAUSE_HOLDER;
			if (hasWaiters()) {
				notifyAll();
			}
		}

		notifyListeners();
		return true;
	}

	@Override
	public boolean isCancelled() {
		return isCancelled0(result);
	}

	private static boolean isCancelled0(Object result) {
		return result instanceof CauseHolder && ((CauseHolder) result).cause instanceof CancellationException;
	}

	@Override
	public boolean isDone() {
		return isDone0(result);
	}

	private static boolean isDone0(Object result) {
		return result != null && result != TaskState.UNCANCELLABLE;
	}
	 @Override
	public boolean setUncancellable() {
		Object result = this.result;
		if (isDone0(result)) {
			return !isCancelled0(result);
		}

		synchronized (this) {
			// Allow only once.
			result = this.result;
			if (isDone0(result)) {
				return !isCancelled0(result);
			}

			this.result = TaskState.UNCANCELLABLE;
		}
		return true;
	}

	@Override
	public V get() throws InterruptedException, ExecutionException {
		await();

		Throwable cause = cause();
		if (cause == null) {
			return getResult();
		}
		throw new ExecutionException(cause);
	}
	@SuppressWarnings("unchecked")
	@Override
	public V getResult() {
		Object result = this.result;
		if (result instanceof CauseHolder || result == TaskState.SUCCESS) {
			return null;
		}
		return (V) result;
	}

	@Override
	public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
		return await0(unit.toNanos(timeout), true);
	}

	@Override
	public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
		if (await(timeout, unit)) {
			Throwable cause = cause();
			if (cause == null) {
				return getResult();
			}
			throw new ExecutionException(cause);
		}
		throw new TimeoutException();
	}

	@Override
	public void addListener(FutureListener<? extends ListenableFuture<? super V>> listener) {

		if (listener == null) {
			throw new NullPointerException("listener");
		}

		if (isDone()) {
			notifyListener(listener);
            return;
		}

		synchronized (this) {
			if (!isDone()) {
				if (listeners != null) {
					log.warn("already add :{} replace it",listeners);
				}
				listeners = listener;
				return;

			}
		}

		notifyListener(listener);

	}

	 
	@SuppressWarnings({ "unchecked", "rawtypes" })
	private void notifyListener(FutureListener listener) {
		try {
			listener.operationComplete(this);
		} catch (Exception e) {
			log.warn("An exception was thrown by {} .operationComplete() error:{}", listeners.getClass(), e);
		}

	}

	@Override
	public ListenableFuture<V> await() throws InterruptedException {
		if (isDone()) {
			return this;
		}

		if (Thread.interrupted()) {
			throw new InterruptedException(toString());
		}

		synchronized (this) {
			while (!isDone()) {

				incWaiters();
				try {
					wait();
				} finally {
					decWaiters();
				}
			}
		}
		return this;
	}

	@Override
	public Throwable cause() {
		Object result = this.result;
		if (result instanceof CauseHolder) {
			return ((CauseHolder) result).cause;
		}
		return null;
	}

	@Override
	public ListenableFuture<V> setFailure(Throwable cause) {
		if (setFailure0(cause)) {
			notifyListeners();
			return this;
		}
		throw new IllegalStateException("complete already: " + this, cause);
	}

	@Override
	public boolean tryFailure(Throwable cause) {
		if (setFailure0(cause)) {
			notifyListeners();
			return true;
		}
		return false;
	}

	@SuppressWarnings("rawtypes")
	private FutureListener listeners;

	private void notifyListeners() {

		@SuppressWarnings("unchecked")
		FutureListener<Future<?>> listeners = this.listeners;
		if (listeners == null) {
			return;
		}
		try {
			listeners.operationComplete(this);
		} catch (Exception e) {
			log.warn("An exception was thrown by {} .operationComplete() error:{}", listeners.getClass(), e);
		}
		 

	}

	private boolean setFailure0(Throwable cause) {
		if (cause == null) {
			throw new NullPointerException("cause");
		}

		if (isDone()) {
			return false;
		}

		synchronized (this) {
			// Allow only once.
			if (isDone()) {
				return false;
			}

			result = new CauseHolder(cause);
			if (hasWaiters()) {
				notifyAll();
			}
		}
		return true;
	}

	@Override
	public ListenableFuture<V> setSuccess(final V result) {
		if (setSuccess0(result)) {
			notifyListeners();
			return this;
		}
		throw new IllegalStateException("complete already: " + this);
	}

	private boolean setSuccess0(final V result) {
		if (isDone()) {
			return false;
		}

		synchronized (this) {
			// Allow only once.
			if (isDone()) {
				return false;
			}
			if (result == null) {
				this.result = TaskState.SUCCESS;
			} else {
				this.result = result;
			}
			if (hasWaiters()) {
				notifyAll();
			}
		}
		return true;
	}
	
    @Override
    public boolean isSuccess() {
        Object result = this.result;
        if (result == null || result == TaskState.UNCANCELLABLE) {
            return false;
        }
        return !(result instanceof CauseHolder);
    }
    

	private static final class CauseHolder {
		final Throwable cause;

		CauseHolder(Throwable cause) {
			this.cause = cause;
		}
	}
}
